<?php
/**
 * SignalsyMQ
 */

 class Signalsy_MQ  
 {
 	//дефолтные настройки
 	private $_options = Array(
 		'maxQueues' => 100000, //максимальное количество очередей
 		'maxMessagesPerQueues' => 1000, //максимальное количество сообщений в одной очереди
 		'expireQueues' => 15552000, //максимальнео время жизни очереди - 180 дней
 		'expireMessages' => 604800, //время жизни максимальное одного сообщения
 		'maxMessageLength' => 32768, // максимальный размер в байтах сообщения (после сериализации)
 		'log' => false, //если указан, то обьект Zend_Log куда логгировать сообщения,
 	    'logMessages' => false, //записывать ли сообщения в лог
 		'redis' => Array(
					 		'namespace' => 'smq_', //неймспейс для ключей, для сокращения пмяти выбран небольшим
					 		'servers' => array(
					 							array('host'     => 'localhost', // Host
										              'port'     => 6379, // Port
										              'weight'   => 1, // Weight of server for key distribution
										              'password' => '' // Password
										        )/*, //дефолтный сервер, может быть несколько
										        array('host'     => 'localhost', // Host
										              'port'     => 6380, // Port
										              'weight'   => 2, // Weight of server for key distribution
										              'password' => '' // Password
										        )*/
					 		),
					 		'keyDistributor' => 'crc32',  //тип распределения ключей
					 		'serializer' => array('Signalsy_MQ', '_serialize'),
							'unserializer' => array('Signalsy_MQ', '_unserialize'))
 	);
 	
 	private $options = null; //текущие настройки
 	
 	//инстанс редиски
 	private $_rediska = null;
 	
 	private $_log = null;
 	
 	private static $_instance = null; //инстанс
 	
 	private $_patterns = array('?', '*'); // apttern symbols at queue name, 
 	
 	//дефолтное сообщение в очеред при ее сздании
 	private $_creatingMQMessages = array('type' => 'init', 'channel' => null, 'body' => null, 'timestamp' => null);
 	
 	
 	/**
 	 * 
 	 * @param unknown_type $config
 	 * @return unknown_type
 	 */
 	static public function getInstance($config = null)
 	{
 		if (self::$_instance === null)
 		{
 			try
 			{
 				//trying find in Zend_Registry
 				if (Zend_Registry::isRegistered('SignalsyMQ') === false)
 				{
 					self::$_instance = new Signalsy_MQ($config);
 				}
 				else
 				{
 					self::$_instance = Zend_Registry::getInstance()->get('SignalsyMQ');
 				} 				
 				
 				if ((is_object(self::$_instance)) && (self::$_instance instanceof Signalsy_MQ))
 				{
 					Zend_Registry::getInstance()->set('SignalsyMQ', self::$_instance);
 					
 					return self::$_instance;
 				}
 				else
 					throw new Signalsy_MQ_Exception('Bad instance creating');
 			}
 			catch (Exception $e)
 			{
 				var_dump($e);
 				return false;
 			}
 		}
 		else
 			return self::$_instance;
 	}
 	
 	//служебный метод чтобы получить редиску :)
 	public function getRediska()
 	{
 		return $this->_rediska;
 	}
 	
 	//конструктор
 	private function __construct($options = null)
 	{
 		if (!empty($options))
 		{
 			$this->options = array_merge($this->_options, $options);
 		}
 		else
 			$this->options = $this->_options;
 		
 		//создаем лог
 		if ((!empty($this->options['log'])) && ($this->options['log'] instanceof Zend_Log))
 		{
 			$this->_log = $this->options['log'];
 		}
 		
 		try
 		{
 			$this->_rediska = new Rediska($this->options['redis']);
 		}
 		catch (Rediska_Exception $e)
 		{
 			throw new Signalsy_MQ_Exception($e->getMessage());
 		} 		
 	}
 	
 	
 	/**
 	 * Setup logger (instance of Zend_Log
 	 * 
 	 * @param $logger
 	 * @return Boolean
 	 */
 	public function setLog(Zend_Log $logger = null)
 	{
 		if ((!empty($logger)) && ($logger instanceof Zend_Log))
 		{
 			$this->_log = $logger;
 			return true;
 		}
 		else
 			return false;
 	}
 	
 	//логгирует событие
 	private function _log($msg = null, $type = Zend_Log::INFO)
 	{
 		if ((!empty($msg)) && (isset($this->_log)) && ($this->_log instanceof Zend_Log))
 		{
 			$this->_log->log($msg, $type);
 		}	
 	}
 	
 	/* Setup serializer function (fn name or array (CLASS, METHOD), see user_function_call)
 	 * 
 	 * @param string|Array $callBack
 	 * @return Boolean
 	 *  
 	 */
 	public function setSerializer($callBack = null)
 	{
 		if (!empty($callBack))
 		{
 			try
 			{
 				return $this->_rediska->setSerializer($callBack);
 			}
 			catch (Rediska_Exception $e)
 			{
 				//setup default
 				return $this->_rediska->setSerializer($this->_options['redis']['serializer']);
 			}
 		}
 		else
 			return false;
 	}
 	
 	
    /* Setup unserializer function (fn name or array (CLASS, METHOD), see user_function_call)
 	 * 
 	 * @param string|Array $callBack
 	 * @return Boolean
 	 *  
 	 */
 	public function setUnserializer($callBack = null)
 	{
 		if (!empty($callBack))
 		{
 			try
 			{
 				return $this->_rediska->setUnserializer($callBack);
 			}
 			catch (Rediska_Exception $e)
 			{
 				//setup default
 				return $this->_rediska->setUnserializer($this->_options['redis']['unserializer']);
 			}
 		}
 		else
 			return false;
 	}
 	
 	
 	
 	
 	//получить канонизированое имя ключа (очереди)
 	private function _getQueueName($name = null)
 	{
 		if (empty($name)) return false;
 		else
 		{
 			$name = htmlentities(strtolower(trim($name)), ENT_NOQUOTES, 'UTF-8', false);
 			
 			if (strlen($name) > 100)
 			{
 				$name = substr($name, 0, 100);
 			}
 			
 			//замена спец-символов
 			$name = str_replace(array('?','*', '&', '|', '\'', '"', '@', '~', '`', '$', '%', '^', '(', ')', '{', '}', '[', ']', '+', '-', '=', ',', ';', ':', '>', '<', '#', '№', ' '), '', $name);
 			
 			return $name; //namespace добавляет сам редис
 		}
 	}
 	
    
 	/**
     * Returns the configuration options for the queue
     *
     * @return array
     */
    public function getOptions()
    {
        return $this->options;
    }
 	
 	//default serializer (Zend_Json)
 	static public function _serialize($data = null)
 	{
 		return Zend_Json::encode($data, true);
 	}
 	
 	//default unserializer (Zend_Json)
 	static public function _unserialize($data = null)
 	{
 		return Zend_Json::decode($data, Zend_Json::TYPE_ARRAY);
 	}
 	
 	/**
 	 * Create new queue
 	 * 
 	 * @param String $name
 	 * @return Boolean
 	 */
 	public function createQueue($name = null)
 	{
 		if ((!empty($name)) && (is_string($name)))
 		{
 			//getting canonize queue name
 			$name = $this->_getQueueName($name);
			
 			if ($name !== false)
 			{
 				//проверяем, есть ли такая очеред в Redis-е, а также тип
 				$check_res = $this->isExists($name);
 				
 				if ($check_res === false) 
 				{
 					$this->_creatingMQMessages['timestamp'] = time();
 						
 					$this->_creatingMQMessages['channel'] = $name;
 						
 					$this->_rediska->prependToList($name, $this->_creatingMQMessages);
 					
 					$this->_log('['.$name.'] Creating queue');
					
					return true; 
  				}				 				
 			}						
 		}
 		
 		return false;
 	}
 	
 	/**
 	 * Проверка существования очереди и ее типа
 	 * 
 	 * @param String $name
 	 * @return Boolean
 	 */
 	public function isExists($name = null)
 	{
 		if (empty($name)) return false;
 		else
 		{
 			$name = $this->_getQueueName($name);
 			
 			if ($name !== false)
 			{
 				//проверим, есть ли такая очередь
 				$check_res = $this->_rediska->exists($name);
 				
 				if ($check_res === false) return false;
 				else
 				{
 					//проверим тип
 					$res_type = $this->_rediska->getType($name);
					
 					if ($res_type != 'list') return false;
 					else
 						return true; 						
 				}
 			}
 		}
 		
 		return false;
 	}
 	
 	
 	/**
     * Return the approximate number of messages in the queue
     *
     * @param  String $queue
     * @return integer
     */
    public function count($queue = null)
    {
        $name = $this->_getQueueName($queue);
        
        if ($name === false) return false;
         
    	if ($this->isExists($name))
        {
        	return $this->_rediska->getListLength($name);
        }
        else
        	return false;    	
    }
    
    
    /**
     * Delete the queue     
     *
     * @return boolean
     */
    public function deleteQueue($queue = null)
    {
    	$name = $this->_getQueueName($queue);
        
        if ($name === false) return false;
        
    	if ($this->isExists($name))
        {
        	$this->_rediska->delete($name);
        	
        	$this->_log('['.$name.'] Deleting queue');
        	
        	return true;
        }
        
        return true; 	
    }
 	
 	
 	/**
     * Send a message to the queue
     * 
     * messages['channel'] = содержит указание куда в какую очередь положить сообщени
     *
     * @param  Array $message
     * @param integer|boolean  skipCheckQueueLength - true/false или число 0 - 100, вероятность проверки
     * @return Boolean
     */
    public function send($message = null, $skipCheckQueueLength = true, $skipCheckQueueExists = true)
    {
    	$isM = $this->_checkMessageFormat($message);
    	if ($isM === false) return false;
    	
    	$queues = Array(); //массив очередей, куда доставляем сообщение
    	
    	$message['channel'] = strtolower($message['channel']);
    	
    	$tmp = $this->_checkQueueNameByPattern($message['channel']);
    	
    	if ($tmp === true)
    	{
    		$queues = $this->getQueuesByPattern($message['channel']);
    	}
    	else
    		$queues[] = $this->_getQueueName($message['channel']);	
    		
		
   		if (empty($queues)) return false;
   		else
   		{
   			try
   			{
   				foreach ($queues as $q_name)
    			{
    				$message['channel'] = strtolower($q_name);
    				
    				$this->_send($message, $skipCheckQueueLength, $skipCheckQueueExists);
    			}
    			
    			return true;
   			}
   			catch (Signalsy_MQ_Exception $e)
   			{
   				return false;
   			}
   		}    	
    }
    
    
    
    /*
     * Sending message to queue
     * @return Boolean
     * @throw Signalsy_MQ_Exception
     */
    private function _send($message = null, $skipCheckQueueLength = true, $skipCheckQueueExists = true)
    {
        $name = $message['channel'];
        
        if (($name === false) || (empty($name))) return false;
        else
        {
        	//проверяем длину очереди
        	if (((is_bool($skipCheckQueueLength)) && ($skipCheckQueueLength === false)) || 
        	   ( (is_int($skipCheckQueueLength)) && (rand(0, 100) < $skipCheckQueueLength)))
        	{
	        	if ($this->_rediska->getListLength($name) > $this->options['maxMessagesPerQueues'])
	        	{
	        		//усечь очередь
	        		$this->_rediska->truncateList($name, $this->options['maxMessagesPerQueues'], 0);
	        		
	        		$this->_log('['.$name.'] Truncate lagester queue', Zend_Log::NOTICE);
	        	}
        	}
        	
        	//проверяем существование очереди
        	if ($skipCheckQueueExists !== true)
        	{
        		$this->_log('['.$name.'] Queue does not exists', Zend_Log::ERR);
        		
        		if (!$this->isExists($name)) throw new Signalsy_MQ_Exception('Queue does not exists');
        	}
        	        	
        	//вставляем
        	if ($this->_rediska->appendToList($name, $message))
        	{
        		if ($this->options['logMessages'] === true)
        		{
        			$this->_log('['.$name.'] Messages: ' . $this->_serialize($message));
        		}
        		
        		return true;
        	} 
        	else        	
        		return false;
        }
    }
    
    
    //служебная функция, проверяет корректность формата сообщения (JSON/Array)
    private function _checkMessageFormat($message = null)
    {
    	if ((!is_array($message)) || 
    	   (empty($message)) || 
    	   (!isset($message['channel'])) || 
    	   (empty($message['channel'])) || 
    	   (!isset($message['timestamp'])) || 
    	   (empty($message['timestamp'])) || 
    	   (!isset($message['body'])) || 
    	   (!isset($message['type'])) ||  (empty($message['type'])))
    	   {
    	   		return false;
    	   }
    	   else
    	   		return true;  
    	   	//!TODO: сделать проверку на длину сообщения  	   
    }
    
    
    /**
     * Return the first element in the queue
     *
     * @param  integer $maxMessages
     * @param  integer $afterTimestamp - указывает на то, что надо проверить возвращенный результат и отдать только сообщения после этой даты
     * @return Zend_Queue_Message_Iterator
     */
    public function receive($queue = null, $maxMessages=null, $afterTimestamp = null)
    {
    	$name = $this->_getQueueName($queue);
        
        if ($name === false) return false;
        
    	if ($this->isExists($name))
    	{
    		//получим указанное количество сообщений
    		$howMatch = $this->options['maxMessagesPerQueues'];
    		
    		if ((!empty($maxMessages)) && ($maxMessages < $howMatch))  $howMatch = $maxMessages;
    		
    		$_res = Array(); //результат
    		
    		$_res = $this->_rediska->getList($name, $howMatch, 0); //от начала
    		
    		$this->_log('['.$name.'] Getting messages: max receive:  ' . $howMatch . '');
    		
    		if ((!is_array($_res)) || (count($_res) < 1)) return false;
    		else
    		{
    			//значения уже ансериализированы
    			if (($afterTimestamp === null) || ($afterTimestamp === false)) 
    			return $_res;
    			else
    			{
    				$_end = Array();
    				foreach ($_res as $p=>$i)
    				{
    					if ((isset($i['timestamp'])) && ($i['type'] != 'init') && ($i['timestamp'] >= $afterTimestamp))
    					{
    						$_end[$p] = $i; //!Важно, позиция это индекс в очереди
    					}
    				}
    				
    				if (!empty($_end))
    				{
    					$this->_log('['.$name.'] Getting messages: returned:  ' . count($_end) . ' messages');
    					
    					return $_end;
    				} 
    				else
    					return false;
    			}
    		}    		
    	}
    }
    
    /**
     * Deleting messages from your queue (defined by channel field)
     * @param string $message
     * @return boolean
     */
    public function deleteMessage($message = null)
    {
    	$name = $message['channel'];
    	
    	//delete only from one queue
    	if ($this->_checkQueueNameByPattern($name) === false)
    	{
    		$name = $this->_getQueueName($name);
    	
    		if (($name === false) || (empty($messages)) || ($this->_checkMessageFormat($messages) == false) ) return false;
    	
    		//удалим из листа
        	$this->_rediska->deleteFromList($name, $messages, 1);
        
        	$this->_log('['.$name.'] Deleting message:' . $this->_serialize($messages));
        
        	return true; 
    	}
    	
    	return false;
    }
    
    /*
     * Delete all messages from queue
     * @param string $queue queue name
     * @return boolean 
     */
    public function clearQueue($queue = null)
    {
    	if (empty($queue)) return false;
    	else
    	{
    		$this->_rediska->truncateList($name, 1);
    		
    		return true;
    	}
    }
    
    /**
     * Getting queue name by pattern ('*' - all symbols, '?' - one symbol) or all (by default)
     * @param string $pattern pattern
     * @return array|boolean
     */
    public function getQueues($pattern = '*')
    {
    	if ($this->_checkQueueNameByPattern($pattern) === true)
    	{
    		$tmp = $this->_rediska->getKeysByPattern($pattern);
    		
    		if (is_array($tmp)) return $tmp;
    	}
    	
    	return false;
    }
    
    /*
     * If queue name (channel) has present pattern symbols, split hea to queue array
     */
    private function _checkQueueNameByPattern($name = null)
    {
    	if (empty($name)) return false;
    	else
    	{    		
    		foreach ($this->_patterns as $pat)
    		{
    			if ((strpos($name, $pat)) !== false) return true; else continue;    			
    		}
    		
    		//check '[' .. ']'
    		
    		if ((strpos($name, '[') !== false) && (strpos($name, ']') !== false)) 
    			return true;
    	}
    	
    	return true;
    }
    
    /**
     * Init manually save Redis DB to disk (async)
     * @return boolean
     */
    public function save()
    {
    	try
    	{
    		$this->_rediska->save(true);
    	}
    	catch (Rediska_Exception $e)
    	{
    		//Исключение может быть в случае, если предыдущее сохранение не завершено
    		return false;
    	}
    	
    	return true;
    }
    
    
    
    

 }